package com.hqq.concurrency.reactor.framework;

import com.sun.org.apache.bcel.internal.generic.Select;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.ServerSocket;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * NIOReactor
 * Created by heqianqian on 2017/8/5.
 */
public class NIOReactor {
    private static final Logger LOGGER = LoggerFactory.getLogger(NIOReactor.class);

    private final Selector selector;
    private final Dispatcher dispatcher;

    private final Queue<Runnable> pendingCommands = new ConcurrentLinkedQueue<>();
    private final ExecutorService reactorMain = Executors.newSingleThreadExecutor();

    public NIOReactor(Dispatcher dispatcher) throws IOException {
        this.dispatcher = dispatcher;
        this.selector = Selector.open();
    }

    public void start() {
        reactorMain.execute(new Runnable() {
            @Override
            public void run() {
                LOGGER.info("Reactor started,waiting for events...");
                try {
                    eventLoop();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    public void stop() throws InterruptedException, IOException {
        reactorMain.shutdownNow();
        selector.wakeup();
        reactorMain.awaitTermination(5, TimeUnit.SECONDS);
        selector.close();
    }

    public NIOReactor registerChannel(AbstractNIOChannel channel) throws ClosedChannelException {
        SelectionKey key = channel.getJavaChannel().register(selector, channel.getInterestedOps());
        key.attach(channel);
        channel.setReactor(this);
        return this;
    }

    private void eventLoop() throws IOException {
        while (true) {
            if (Thread.interrupted()) {//响应中断请求
                break;
            }

            processPendingCommands();

            selector.select();

            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                if (!key.isValid()) {
                    iterator.remove();
                    continue;
                }
                processKey(key);
            }
            selectionKeys.clear();
        }
    }

    private void processPendingCommands() {
        Iterator<Runnable> iterator = pendingCommands.iterator();
        while (iterator.hasNext()) {
            Runnable command = iterator.next();
            command.run();
            iterator.remove();
        }
    }

    private void processKey(SelectionKey key) throws IOException {
        if (key.isAcceptable()) {
            onChannelAcceptable(key);
        } else if (key.isReadable()) {
            onChannelReadable(key);
        } else if (key.isWritable()) {
            onChannelWritable(key);
        }
    }

    private void onChannelWritable(SelectionKey key) throws IOException {
        AbstractNIOChannel channel = (AbstractNIOChannel) key.attachment();
        channel.flush(key);
    }

    private void onChannelReadable(SelectionKey key) {
        try {
            Object readObject = ((AbstractNIOChannel) key.attachment()).read(key);
            dispatchReadEvent(key, readObject);
        } catch (IOException e) {
            try {
                key.channel().close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }
    }

    private void dispatchReadEvent(SelectionKey key, Object readObject) {
        dispatcher.onChannelReadEvent((AbstractNIOChannel) key.attachment(), readObject, key);
    }

    private void onChannelAcceptable(SelectionKey key) throws IOException {
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
        SocketChannel socketChannel = serverSocketChannel.accept();
        socketChannel.configureBlocking(false);
        SelectionKey readKey = socketChannel.register(selector, SelectionKey.OP_READ);
        readKey.attach(key.attachment());
    }

    public void changeOps(SelectionKey key, int interestedOps) {
        pendingCommands.add(new ChangeKeyOpsCommand(key, interestedOps));
    }

    class ChangeKeyOpsCommand implements Runnable {

        private SelectionKey key;
        private int interestedOps;

        public ChangeKeyOpsCommand(SelectionKey key, int interestedOps) {
            this.key = key;
            this.interestedOps = interestedOps;
        }

        @Override
        public void run() {
            key.interestOps(interestedOps);
        }

        @Override
        public String toString() {
            return "Change on ops to: " + interestedOps;
        }
    }
}
